package com.burges.net.assembly.gelly

import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.graph.scala.Graph
import org.apache.flink.graph._
import org.apache.flink.graph.validation.InvalidVertexIdsValidator
import org.apache.flink.types.NullValue

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/26 
  * 描述     Graph API主要操作接口
  */
object GraphAPIDEMO {

	def main(args: Array[String]): Unit = {
		/**
		  * Graph创建
		  */
		val environment = ExecutionEnvironment.getExecutionEnvironment

		//1.从DataSet创建Graph
		// 构建顶点集合
		val vertices: DataSet[Vertex[String, Long]] = environment.fromElements(new Vertex("1", 100L), new Vertex("2", 200L))
		// 构建边的结合
		val edges: DataSet[Edge[String, Long]] = environment.fromElements(new Edge("1", "2", 10L))
		// 将边和顶点通过Graph.fromDataSet方法整合，创建graph
		val graphFromDataSet: Graph[String, Long, Long] = Graph.fromDataSet(vertices, edges, environment)

		//2.从Collection中构建
		// 创建顶点
		val verticesFromList: List[Vertex[Long, Long]] = List[Vertex[Long, Long]](new Vertex[Long, Long](6L, 6L))
		// 创建边
		val edgesFromList: List[Edge[Long, Long]] = List[Edge[Long, Long]](new Edge[Long, Long](6L, 1L, 61L))
		// 创建图
		val graphFromCollection = Graph.fromCollection(verticesFromList, edgesFromList, environment)

		//3.通过从Csv文件中创建Graph
		// 从Csv文件中创建Graph，指定Vertices和Edges路径
		val graphFromCsv = Graph.fromCsvReader[String, Long, Double](
			pathVertices = "path/vertex_file",
			pathEdges = "path/edge_file",
			env = environment)
		// 从Csv文件中创建Graph，不指定Vertices路径
		val simpleGraph = Graph.fromCsvReader[Long, NullValue, NullValue](
			pathEdges = "path/vertex_file",
			env = environment
		)

		/**
		  * Graph转换操作
		  */
		//Map操作
		// 使用mapVertices对vertices的value进行转换
		val updatedGraph = graphFromDataSet.mapVertices( v => v.getValue * 10)
		//Translate操作
		// 将Graph中的ID转换为String类型
		val updatedGraph = graphFromDataSet.translateGraphIds(id => id.toString)
		//Filter操作
		// 过滤Edge的value大于10的Graph
		graphFromDataSet.filterOnEdges( edge => edge.getValue > 10)
		// 过滤Vertices的Value小于10的Graph
		graphFromDataSet.filterOnVertices(vertex => vertex.getValue < 10)
		// 将顶点值为正数和边值为负数的子图筛选出来
		graphFromDataSet.subgraph((vertex => vertex.getValue > 0), (edge => edge.getValue < 0))
		//Join操作
		val dataSet: DataSet[(String, Long)] = environment.fromElements(("3", 300L))
		graphFromDataSet.joinWithEdgesOnSource(dataSet, (v1: Long, v2: Long) => v1+v2)
		//Reverse操作
		val reverseGraph = graphFromDataSet.reverse()
		//Undirected
		val undirectedGraph: Graph[String, Long, Long] = graphFromDataSet.getUndirected()
		//Union操作
		// 注意：两个图的结构完全相同
		val result: Graph[String, Long, Long] = graphFromDataSet.union(graphFromDataSet)
		//Differenct操作
		val differenceGraph: Graph[String, Long, Long] = graphFromDataSet.difference(graphFromDataSet)
		//Intersect操作
		// 注意：第二个参数设定为true，去除重复的Edges，设定为False，不去除重复的Edges
		val intersectGraph: Graph[String, NullValue, Long] = graphFromDataSet.intersect(graphFromDataSet, true)

		/**
		  * 图突变主要是对图的结构进行修改的方法
		  */

		/**
		  * 邻方法：基于顶点对每个顶点的邻近点或者边进行聚合计算
		  * 以下只是示例代码
		  */
		val maxWeights = graphFromDataSet.reduceOnEdges(new SelectMaxWeightFunction, EdgeDirection.OUT)
		val verticesWithSum = graphFromDataSet.reduceOnNeighbors(new SumValuesFunction, EdgeDirection.IN)

		/**
		  * 图校验
		  */
		val verticesIds: DataSet[Vertex[Long, Long]] = environment.fromElements(new Vertex(100L, 100L), new Vertex(200L, 200L))
		val edges:List[Edge[Long, Long]] = List[Edge[Long, Long]](new Edge[Long, Long](1L, 1L, lL), new Edge[Long, Long](2L, 2L, lL), new Edge[Long, Long](3L, 3L, lL))
		val graph = Graph.fromCollection(verticesIds, edges, environment)
		//检验图是否有效，返回ID为4的节点不合法
		graph.validate(new InvalidVertexIdsValidator[Long, Long, Long])


	}

	// 定义ReduceEdgesFunction实现对出度边的权重求和
	final class SelectMaxWeightFunction extends ReduceEdgesFunction[Long]{
		override def reduceEdges(firstEdgeValue: Long, secondEdgeValue: Long): Long = {
			Math.max(firstEdgeValue, secondEdgeValue)
		}
	}

	final  class SumValuesFunction extends ReduceNeighborsFunction[Long]{
		override def reduceNeighbors(firstNeighbor: Long, secondNeighbor: Long): Long = {
			firstNeighbor + secondNeighbor
		}
	}

}
